package com.example.kafkademo.offsetcommit;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * @author yanyun zhu
 * @version 1.0
 * @date 2021/12/29 9:17
 *  处理一段消息，提交一次offset
 */
public class BatchCommit {

    //TopicPartition，即消费的分区，而值是一个 OffsetAndMetadata 对象，保存的主要是位移数据。
    private static Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();


    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "false");    //默认true, 自动提交
//        props.put("auto.commit.interval.ms", "2000");     //表明 Kafka 每 5 秒会为你自动提交一次位移(默认)
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test-a"));

        int count = 0;
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> record : records) {
//                process(record);  // 处理消息
                offsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1));
                if (count % 100 == 0) {
                    consumer.commitAsync(offsets, new OffsetCommitCallback() {
                        @Override
                        public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                            // 回调处理逻辑是
                        }
                    });
                }
                count++;
            }
        }
    }
}
